package com.hskn.hss.config;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.hskn.hss.core.constant.Constants;
import com.hskn.hss.module.dturecord.entity.DtuRecord;
import com.hskn.hss.module.dturecord.mapper.DtuRecordMapper;
import com.hskn.hss.module.kfkanalyse.service.IkfkAnalyseService;
import com.hskn.hss.module.kfkcarvideo.mapper.KfkCarVideoMapper;
import com.hskn.hss.module.kfkdeal.service.impl.KfkHandleServiceImpl;
import com.hskn.hss.module.kfkdtu.entity.KfkDtu;
import com.hskn.hss.module.kfkdtu.mapper.KfkDtuMapper;
import com.hskn.hss.module.kfkhumancamer.mapper.KfkHumanCamerMapper;
import com.hskn.hss.module.kfkinterphone.mapper.KfkInterphoneMapper;
import com.hskn.hss.module.kfknvr.mapper.KfkNvrMapper;
import com.hskn.hss.module.kfkpunchcard.mapper.KfkPunchCardMapper;
import com.hskn.hss.module.kfkworkcard.mapper.KfkWorkCardMapper;
import com.hskn.hss.module.tlemployee.mapper.TlEmployeeMapper;
import com.hskn.hss.utils.JedisPoolUtils;
import com.hskn.hss.utils.KafkaUtils;
import com.hskn.hss.utils.StringUtils;
import com.hskn.hss.utils.redis.RedisUtil;
import com.hskn.hss.utils.redis.key.EquipOnlineKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Date;
import java.util.Map;
import java.util.Objects;

/**
 * @author lny
 * createTime 2021-09-09
 **/
@Order(value = 4)
@Slf4j
@Component
public class KfkDtuRunner implements CommandLineRunner {

    @Autowired
    private KfkHandleServiceImpl kfkHandleService;
    @Resource
    private DtuRecordMapper dtuRecordMapper;
    @Autowired
    TlEmployeeMapper tlEmployeeMapper;
    @Autowired
    KfkWorkCardMapper kfkWorkCardMapper;
    @Autowired
    IkfkAnalyseService ikfkAnalyseService;
    @Autowired
    KfkInterphoneMapper kfkInterphoneMapper;
    @Autowired
    KfkPunchCardMapper kfkPunchCardMapper;
    @Autowired
    KfkHumanCamerMapper kfkHumanCamerMapper;
    @Autowired
    KfkDtuMapper kfkDtuMapper;
    @Autowired
    KfkCarVideoMapper kfkCarVideoMapper;
    @Autowired
    KfkNvrMapper kfkNvrMapper;

    @Value("${kafka.dtu_topic}")
    private String dtu_topic;

    @Value("${kafka.dtu_group}")
    private String dtu_group;

    @Value("${kafka.kafka_receive:true}")
    private Boolean kafka_receive;

    @Override
    public void run(String... args) {
        if (!kafka_receive) {
            return;
        }
        new Thread(() -> {
            KafkaConsumer consumer = KafkaUtils.getConsumer(dtu_group);
            consumer.subscribe(Arrays.asList(dtu_topic));
            while (true) {
                receiveAndHandle(consumer);
            }
        }).start();
    }

    public void receiveAndHandle(KafkaConsumer consumer) {
        //拉取kafka数据
        log.info("====================kfk定时读取任务开始-DTU======================================");
        Jedis jedis = null;
        try {
            jedis = JedisPoolUtils.getJedis();
            ConsumerRecords<Object, Object> records = consumer.poll(1000);
            log.info("========KfkDtuRunner拉取条数：{}", records.count());
            for (ConsumerRecord<Object, Object> record : records) {
                try {
                    Map value = JSONObject.parseObject(String.valueOf(record.value()), Map.class);
                    Map content = JSONObject.parseObject(String.valueOf(value.get("content")), Map.class);
                    Map info = JSONObject.parseObject(String.valueOf(content.get("info")), Map.class);
                    Date timestamp = new Date(record.timestamp());

                    if (value.get("deviceType").equals("DTU") && content.get("type").equals("DTU_REPORT")) { //DTU
                        long dtuStartTime = System.currentTimeMillis();
                        KfkDtu kfkDtu = new KfkDtu();
                        kfkDtu.setEqpnum(String.valueOf(value.get("no")));
                        kfkDtu.setState(String.valueOf(value.get("state")));
                        kfkDtu.setTimestamp(timestamp);
                        kfkDtu.setCategory(String.valueOf(info.get("category")));
                        kfkDtu.setValue(String.valueOf(info.get("value")));
                        kfkDtu.setDeviceTime(DateUtil.parseDateTime(String.valueOf(info.get("createTime"))));
                        kfkDtu.setCreateTime(new Date());
                        kfkDtuMapper.insert(kfkDtu);

                        value.put("locationDate", timestamp);
                        Map map = kfkHandleService.DTULocationHandle(value);

                        //kfk推送的设施记录保存数据库
                        DtuRecord dtuRecord = new DtuRecord();
                        dtuRecord.setEqpnum(String.valueOf(value.get("no")));
                        dtuRecord.setState(Integer.parseInt(String.valueOf(value.get("state"))));
                        dtuRecord.setCategory(String.valueOf(info.get("category")));
                        String valueResult = String.valueOf(info.get("value"));
                        if (!StringUtils.isEmpty(valueResult) && valueResult.length() > 3) {
                            dtuRecord.setValue(valueResult.replace(valueResult.substring(valueResult.length() - 3), ""));
                        }
                        dtuRecord.setTimestamp(timestamp);
                        dtuRecord.setThresholdunit("ppm");
                        if (Objects.nonNull(map.get("is_ammonia_alarm")) && "1".equals(String.valueOf(map.get("is_ammonia_alarm")))) {
                            dtuRecord.setIsAmmoniaAlarm(1);
                        }
                        if (Objects.nonNull(map.get("is_hydrogen_sulfide_alarm")) && "1".equals(String.valueOf(map.get("is_hydrogen_sulfide_alarm")))) {
                            dtuRecord.setIsHydrogenSulfideAlarm(1);
                        }
                        if (Objects.nonNull(map.get("is_smoke_alarm")) && "1".equals(String.valueOf(map.get("is_smoke_alarm")))) {
                            dtuRecord.setIsSmokeAlarm(1);
                        }
                        dtuRecordMapper.insert(dtuRecord);

                        String redisKey = Constants.ABNORMAL + dtuRecord.getEqpnum();
                        jedis.setex(redisKey, StringUtils.getSecondsNextEarlyMorning().intValue(), JSONObject.toJSONString(dtuRecord));

                        //保存设备在线状态
                        RedisUtil.set(EquipOnlineKey.token, kfkDtu.getEqpnum(), new JSONObject());

                        long dtuEndTime = System.currentTimeMillis();
                        log.info("=====DTU=执行时间(毫秒)：{}", dtuEndTime - dtuStartTime);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
            log.info("====================kfk定时读取任务结束-DTU======================================");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            consumer.commitAsync();
            if (jedis != null) {
                jedis.close();
            }
        }
    }
}